package com.mxnavi5.example.DataStream;

import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class TestKeyedProcessFunction extends KeyedProcessFunction <Integer, Tuple2<Integer, Integer>, Integer>  {

    private ValueState<Integer> valueState;
    private ListState<Integer> listState;
    private ReducingState<Integer> reduceState;
    private AggregatingState<Integer,Integer> aggregatingState;



    @Override
    public void open(Configuration parameters) throws Exception {
        //设置状态的超时时间以及更新时间的方式  处理时间24*7小时未进行读写则认为超过设置的生命周期，如果状态未被清空则返回
        StateTtlConfig stateTtlConfig = new StateTtlConfig
                .Builder(Time.hours(24 * 7))
                .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
                .setTtlTimeCharacteristic(StateTtlConfig.TtlTimeCharacteristic.ProcessingTime)
                .build();

        ValueStateDescriptor<Integer> valueStateDescriptor =
                new ValueStateDescriptor<>("valueStateDescriptor", Integer.class);
        valueStateDescriptor.enableTimeToLive(stateTtlConfig);
        valueState = getRuntimeContext().getState(valueStateDescriptor);

        ListStateDescriptor<Integer> listStateDescriptor =
                new ListStateDescriptor<>("listStateDescriptor", Integer.class);
        listStateDescriptor.enableTimeToLive(stateTtlConfig);
        listState = getRuntimeContext().getListState(listStateDescriptor);


        ReducingStateDescriptor<Integer> reduceStateDescriptor =
                new ReducingStateDescriptor<>("reduceStateDescriptor",new MyReduceFunction(), Integer.class);
        valueStateDescriptor.enableTimeToLive(stateTtlConfig);
        reduceState = getRuntimeContext().getReducingState(reduceStateDescriptor);

        AggregatingStateDescriptor<Integer,Tuple2<Integer,Integer>,Integer> aggStateDescriptor =
                new AggregatingStateDescriptor<Integer,Tuple2<Integer,Integer>,Integer>("aggStateDescriptor",new MyAggFunction(),
                        TypeInformation.of(new TypeHint<Tuple2<Integer,Integer>>() {}));
        valueStateDescriptor.enableTimeToLive(stateTtlConfig);
        aggregatingState = getRuntimeContext().getAggregatingState(aggStateDescriptor);

    }

    @Override
    public void processElement(Tuple2<Integer, Integer> value, KeyedProcessFunction<Integer, Tuple2<Integer, Integer>, Integer>.Context ctx, Collector<Integer> out) throws Exception {
        Integer count = valueState.value();
        if(count==null){
            count=0;
        }
        count+=1;
        valueState.update(count);

        listState.add(value.f1);
        reduceState.add(value.f1);
        aggregatingState.add(value.f1);
//        out.collect(value.f1);

        if(count==5){
            //输出
            String sl = "list:";
            Iterator<Integer> it = listState.get().iterator();
            while (it.hasNext()){
                sl=sl+it.next()+",";
            }
            System.out.println(sl);

            System.out.println("reduce:"+reduceState.get());

            System.out.println("agg:"+aggregatingState.get());

            System.out.println("====================输出完毕====================");
            //清空状态
            valueState.clear();
            listState.clear();
            reduceState.clear();
            aggregatingState.clear();
        }

    }
}


